package com.shujia.flink.cep

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.cep.scala.CEP

object Demo1CEP {
  def main(args: Array[String]): Unit = {
    /**
      * 1,192.168.0.1,a
      * 1,192.168.0.2,a
      * 1,192.168.0.3,b
      * 1,192.168.0.4,b
      *
      * 如果同一个id连续两次出现了fail   报警
      */

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    val ds = env.socketTextStream("node1", 8888)

    val eventDS = ds.map(_.split(","))
      .map(arr => Event(arr(0).toInt, arr(1), arr(2)))

    /**
      * 定义规则
      */

    val pattern = Pattern

      .begin[Event]("start") //第一个事件
      .where(event => event.msg.equals("a")) // 条件
      .times(4)
      .consecutive()

      .next("next") //下一个事件
      .where(event => event.msg.equals("b")) // 条件


    //使用规则在流上进行匹配
    val patternStream = CEP.pattern(eventDS.keyBy(_.id), pattern)


    //如果匹配上  就可以取出匹配成功的数据
    patternStream.select(map => {
      val start = map.getOrElse("start", null)
      val next = map.getOrElse("next", null)

      println(start + "\t" + next)

    })


    env.execute()


  }

  case class Event(id: Int, ip: String, msg: String)

}
